package org.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TrademarkCategoryUserRefundBean;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.function.AsyncDimFunction;
import org.atguigu.gmall.realtime.util.AtguiguUtil;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dws
 * Author: ZengHaiFeng
 * CreateTime:2023/5/5 20:29:09
 * Description: Todo
 * Version：1.0
 */
public class Dws_11_DwsTradeTrademarkCategoryUserRefundWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_11_DwsTradeTrademarkCategoryUserRefundWindow()
                .init(40010,2,
                        Constant.TOPIC_DWD_TRADE_ORDER_REFUND,
                        "Dws_10_DwsTradeTrademarkCategoryUserRefundWindow");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
       // parse data to pojo
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> parseStream = parseToPojo(ds);
        // dim
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> dimensionStream = dimension(parseStream);
        // window and aggregation
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> windowStream = windowAndAgg(dimensionStream);
        // supplementary dimension
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> resultStream = joinDims(windowStream);
        // output to Clickhouse
        writeToClickhouse(resultStream);

    }

    private void writeToClickhouse(SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink(
                "dws_trade_trademark_category_user_refund_window",
                TrademarkCategoryUserRefundBean.class
        ));
    }

    private SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> joinDims(SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> windowStream) {
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> tmStream = AsyncDataStream.unorderedWait(
                windowStream,
                new AsyncDimFunction<TrademarkCategoryUserRefundBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_trademark";
                    }

                    @Override
                    public String getId(TrademarkCategoryUserRefundBean bean) {
                        return bean.getTrademarkId();
                    }

                    @Override
                    public void addDim(TrademarkCategoryUserRefundBean bean, JSONObject dim) {
                        bean.setTrademarkName(dim.getString("TM_NAME"));
                    }
                }
                , 60, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> c3Stream = AsyncDataStream.unorderedWait(
                tmStream,
                new AsyncDimFunction<TrademarkCategoryUserRefundBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category3";
                    }

                    @Override
                    public String getId(TrademarkCategoryUserRefundBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public void addDim(TrademarkCategoryUserRefundBean bean, JSONObject dim) {
                        bean.setCategory3Name(dim.getString("NAME"));
                        bean.setCategory2Id(dim.getString("CATEGORY2_ID"));
                    }
                }
                , 60, TimeUnit.SECONDS);
        SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> c2Stream = AsyncDataStream.unorderedWait(
                c3Stream,
                new AsyncDimFunction<TrademarkCategoryUserRefundBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category2";
                    }

                    @Override
                    public String getId(TrademarkCategoryUserRefundBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public void addDim(TrademarkCategoryUserRefundBean bean, JSONObject dim) {
                        bean.setCategory2Name(dim.getString("NAME"));
                        bean.setCategory1Id(dim.getString("CATEGORY1_ID"));
                    }
                }
                , 60, TimeUnit.SECONDS);
        return AsyncDataStream.unorderedWait(
                c2Stream,
                new AsyncDimFunction<TrademarkCategoryUserRefundBean>() {
                    @Override
                    public String getTable() {
                        return "dim_base_category1";
                    }

                    @Override
                    public String getId(TrademarkCategoryUserRefundBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public void addDim(TrademarkCategoryUserRefundBean bean, JSONObject dim) {
                        bean.setCategory1Name(dim.getString("NAME"));
                    }
                }
                , 60, TimeUnit.SECONDS);
    }

    private SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> windowAndAgg(SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> dimensionStream) {
        return dimensionStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrademarkCategoryUserRefundBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean,ts) -> bean.getTs())
                                .withIdleness(Duration.ofSeconds(60))
                )
                .keyBy(value -> value.getTrademarkId() + value.getCategory3Id() + value.getUserId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TrademarkCategoryUserRefundBean>() {
                    @Override
                    public TrademarkCategoryUserRefundBean reduce(
                            TrademarkCategoryUserRefundBean value1, TrademarkCategoryUserRefundBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        return value1;
                    }
                }, new ProcessWindowFunction<TrademarkCategoryUserRefundBean, TrademarkCategoryUserRefundBean, String, TimeWindow>() {
                    @Override
                    public void process(
                            String s,
                            Context context,
                            Iterable<TrademarkCategoryUserRefundBean> elements,
                            Collector<TrademarkCategoryUserRefundBean> out) throws Exception {
                        TrademarkCategoryUserRefundBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        bean.setRefundCount((long)bean.getOrderIdSet().size());
                        out.collect(bean);
                    }
                });
    }

    private SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> dimension(SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> parseStream) {
        return AsyncDataStream.unorderedWait(
                parseStream, new AsyncDimFunction<TrademarkCategoryUserRefundBean>() {
                    @Override
                    public String getTable() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getId(TrademarkCategoryUserRefundBean bean) {
                        return bean.getSkuId();
                    }

                    @Override
                    public void addDim(TrademarkCategoryUserRefundBean bean, JSONObject dim) {
                        bean.setTrademarkId(dim.getString("TM_ID"));
                        bean.setCategory3Id(dim.getString("CATEGORY3_ID"));
                    }
                }
        , 60, TimeUnit.SECONDS );
    }

    private SingleOutputStreamOperator<TrademarkCategoryUserRefundBean> parseToPojo(DataStreamSource<String> ds) {
        return ds
                .map(value -> {
                    JSONObject obj = JSON.parseObject(value);
                    return TrademarkCategoryUserRefundBean.builder()
                            .orderIdSet(new HashSet<>(Collections.singletonList(obj.getString("order_id"))))
                            .userId(obj.getString("user_id"))
                            .skuId(obj.getString("sku_id"))
                            .ts(obj.getLong("ts")*1000)
                            .build();
                });
    }
}
